package cn.me.alphamq.nameserver.event.spi.listener.cluster.chain;

import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.sync.StopWatch;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.nameserver.enumeration.ReplicationMsgCodeEnum;
import cn.me.alphamq.nameserver.event.model.cluster.ReplicationAckEvent;
import cn.me.alphamq.nameserver.event.model.cluster.ReplicationMsgEvent;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.nameserver.store.ServiceInstance;
import cn.me.alphamq.common.store.StopWatchManager;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 链式复制中非头节点的监听器，监听前一个节点发送过来的复制消息事件
 *
 * @author f
 */
public class NodeReplicationMsgListener implements Listener<ReplicationMsgEvent> {

    private static final Logger log = LoggerFactory.getLogger(NodeReplicationMsgListener.class);

    /**
     * 停表
     */
    private final StopWatch stopWatch = StopWatchManager.newStopWatch();

    @Override
    public void onReceive(ReplicationMsgEvent event) throws Exception {
        // 接收前一个节点同步的数据，并进行对应的修改，并把复制消息放到队列中
        ReplicationMsgCodeEnum msgCodeEnum = ReplicationMsgCodeEnum.of(event.getMsgCode());
        if (ReplicationMsgCodeEnum.INSERT_OR_UPDATE == msgCodeEnum) {
            // 如果是 添加/更新 数据，则使用 put()
            ServiceInstance serviceInstance = event.getServiceInstance();
            // 将主 nameserver 节点与其它服务实例的 channel 的上下文设置为 null
            serviceInstance.setCtx(null);
            Container.getServiceInstanceManager().put(serviceInstance);

            // 将复制消息传递给下一个节点，并且等待下一个节点返回响应，以此唤醒当前线程
            // 如果当前节点是非尾节点，则会先阻塞，然后在下一个节点返回响应时唤醒；如果当前节点是尾节点，则不会阻塞
            Container.getReplicationMsgQueueManager().putShiftyServiceInstance(serviceInstance, stopWatch);

            // 添加/更新 数据需要强一致性，非头节点需要向前一个节点返回 ack
            event.getCtx().writeAndFlush(new TcpMsg(EventCodeEnum.NODE_REPLICATION_ACK_REQ,
                    JSON.toJSONBytes(new ReplicationAckEvent(event.getMsgId()))));
//            log.info("非头节点更新数据成功"); // TODO DELETE
        } else if (ReplicationMsgCodeEnum.DELETE == msgCodeEnum) {
            // 如果是 删除 数据，则使用 remove()
            String key = event.getKey();
            Container.getServiceInstanceManager().remove(key);
            Container.getReplicationMsgQueueManager().putRemovedServiceInstance(key);
        }
    }
}
